package com.shujia.mllib

import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, DataFrameReader, Dataset, Row, SaveMode, SparkSession}

object Demo04ImagePrePro {
  def main(args: Array[String]): Unit = {

    // 1、加载数据 并进行特征工程处理
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo04ImageTrain")
      .master("local[*]")
      .config("spark.sql.shuffle.partitions", "8")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val imageDF: DataFrame = spark
      .read
      .format("image")
      .load("C:\\Users\\zzk10\\Desktop\\Spark\\MLLib\\train")

    imageDF.printSchema()

    //    imageDF.show(10, truncate = false)

    val imageProDF: DataFrame = imageDF
      .select($"image.origin" as "origin", $"image.data" as "data")
      .as[(String, Array[Byte])]
      .map {
        case (filePath: String, data: Array[Byte]) =>
          val fileName: String = filePath.split("/").last
          val iArr: Array[Double] = data.map(i => {
            if (i.toInt < 0) {
              1.0
            } else {
              0.0
            }
          })
          (fileName, Vectors.dense(iArr).toSparse)
      }.toDF("fileName", "features")

    // 读取图片对应的结果数据 然后关联获取label
    val imageResDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("fileName String,label Double")
      .load("Spark/data/mllib/data/image_res.txt")

    val imageDataDF: DataFrame = imageProDF
      .join(imageResDF, "fileName")
      .select("fileName", "features", "label")

    imageDataDF
      .select("label", "features")
      .write
      .format("libsvm")
      .mode(SaveMode.Overwrite)
      .save("Spark/data/mllib/imageFeatures")





  }

}
